package com.test.study.ToolsUtils.strom.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * 过滤的topology
 * 
 * @author huyong
 *
 */
public class TridentTopologyFilter {

	@SuppressWarnings("unchecked")
	public static void main(String[] args) {
		TridentTopology topology = new TridentTopology();
		FixedBatchSpout fixedBatchSpout = new FixedBatchSpout(new Fields("sentence"), 4, new Values(1), new Values(2),
				new Values(3), new Values(4));
		topology.newStream("tupleId", fixedBatchSpout).each(new Fields("sentence"), new MyTridentFilter())
				.each(new Fields("sentence"), new MyTridentFunction(), new Fields(""));
		fixedBatchSpout.setCycle(true);

		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("TridentTopologyFilter", new Config(), topology.build());

	}

}
